package com.haozhen.spark

import akka.actor.{Actor, ActorRef, ActorSystem, Props}
import com.haozhen.spark.Common.{RegisterWorkInfo, RegisteredWorkerInfo, RemoveTimeoutWorker, StartMessage, StartTimeoutCheck, WorkInfo}
import com.typesafe.config.ConfigFactory

import scala.collection.mutable

/**
  * @author haozhen
  * @email haozh@ync1.com
  * @date 2020/12/8  23:24
  *       5、模拟Spark中Master与Worker进程通讯
  *       为了加深对主从服务心跳检测机制（HeartBeat）的理解，模拟master与slave之间的通信。
  *       作业要求：
  *       1. Worker 注册到 Master，Master 完成注册，并回复 Worker 注册成功(注册功能)
  *       2. Worker 定时发送心跳，并在 Master 接收到
  *       3. Master 接收到 Worker 心跳后，要更新该 Worker 的最近一次发送心跳的时间
  *       4. 给 Master 启动定时任务，定时检测注册的 Worker 有哪些没有更新心跳，并将其从 hashmap 中删除
  */

class SparkMaster extends Actor{

  val workers = mutable.HashMap[String,WorkInfo]()



  override def receive: Receive = {
    case StartMessage(msg)=>{
      println(s"Spark Master start. $msg")
      self ! StartTimeoutCheck
    }
    case RegisterWorkInfo(id,cpu,ram)=> {
      if (!workers.contains(id)) {
        val workInfo = new WorkInfo(id, cpu, ram)
        workers += (id -> workInfo)
        sender ! RegisteredWorkerInfo
        println(s"worker[$id]完成注册")
      }
    }
      case StartTimeoutCheck=>{
        import context.dispatcher
        import scala.concurrent.duration._
        println("master开启定时检测心跳任务。。。。")
        context.system.scheduler.schedule(0 millis,10000 millis,self,RemoveTimeoutWorker)
      }
    case  RemoveTimeoutWorker=>{
      println("心跳检测中。。。")
      val current = System.currentTimeMillis()
      val workersInfo = workers.values
      workersInfo.filter(current - _.lasterHeartBeatTime >6000)
        .foreach(workersInfo=>workers.remove(workersInfo.id))
    }

  }
}


object SparkMaster {

  val masterHost = "127.0.0.1"
  val masterPort = 10000
  val config = ConfigFactory.parseString(
    s"""
       |akka.actor.provider="akka.remote.RemoteActorRefProvide"
       |akka.remote.netty.tcp.hostname=$masterHost
       |akka.remote.netty.tcp.port=$masterPort
     """.stripMargin
  )

  val sparkMasterActorSystem = ActorSystem("SparkMaster",config)

  val sparkMaster:ActorRef = sparkMasterActorSystem.actorOf(Props[SparkMaster],"SparkMaster")

  sparkMaster ! StartMessage(s"host:$masterHost port:$masterPort")
}
